package com.alison.tableapisql.chapter2_timeAttr;

import com.alison.tableapisql.chapter1_tableapiandsql.model.SensorReading;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class E2_TableTest5_TimeAndWindow {
    /*
    注：这里最后一列rt里显示的是EventTime，而不是Processing Time
(
  `id` STRING,
  `ts` BIGINT,
  `temp` DOUBLE,
  `rt` TIMESTAMP(3) *ROWTIME*
)

+I[sensor_1, 1547718199, 35.8, 2019-01-17T09:43:19]
+I[sensor_6, 1547718201, 15.4, 2019-01-17T09:43:21]
+I[sensor_7, 1547718202, 6.7, 2019-01-17T09:43:22]
+I[sensor_10, 1547718205, 38.1, 2019-01-17T09:43:25]
+I[sensor_1, 1547718207, 36.3, 2019-01-17T09:43:27]
+I[sensor_1, 1547718209, 32.8, 2019-01-17T09:43:29]
+I[sensor_1, 1547718212, 37.1, 2019-01-17T09:43:32]
     */
    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String filePath = "D:\\workspace\\lab\\learnbigdata\\learnflink\\flink-datastream\\src\\main\\resources\\tableapi\\E1.txt";

        // 2. 读入文件数据，得到DataStream
        DataStream<String> inputStream = env.readTextFile(filePath);

        // 3. 转换成POJO
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
                    String[] fields = line.split(",");
                    return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
                })
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)) {
                    @Override
                    public long extractTimestamp(SensorReading element) {
                        return element.getTimestamp() * 1000L;
                    }
                });
        // 4. 将流转换成表，定义时间特性
        //        Table dataTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp, pt.proctime");
        // 注：这里最后一列rt里显示的是EventTime，而不是Processing Time
        Table dataTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp, rt.rowtime");
        dataTable.printSchema();

        tableEnv.toAppendStream(dataTable, Row.class).print();
        env.execute();
    }
}